package com.chb.flink.table


import com.chb.flink.source.{MyCustomerSource, StationLog}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.{EnvironmentSettings, Slide, Table, Tumble}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row

object TestWindowByTableAPI {

    //每隔5秒钟统计，每个基站的通话数量,假设数据是乱序。最多延迟3秒,需要水位线
    def main(args: Array[String]): Unit = {
        //使用Flink原生的代码创建TableEnvironment
        //先初始化流计算的上下文
        val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        streamEnv.setParallelism(1)
        val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
        val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)


        streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


        // 设置时间语义

        //两个隐式转换
        import org.apache.flink.streaming.api.scala._
        // 第二个隐式转换， Table
        import org.apache.flink.table.api.scala._

        //读取数据源，
        val stream = streamEnv.addSource(new MyCustomerSource)
            // 引入watermark
            .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[StationLog](Time.seconds(3)) {
                override def extractTimestamp(t: StationLog): Long = {
                    t.callTime
                }
            })


        // 动态创建table
        val table = tableEnv.fromDataStream(stream, 'sid, 'callOut, 'callIn, 'callType, 'callTime.rowtime)

        // 开窗 ,  滚动窗口两种写法
        // table.window(Tumble.over("5.second").on("callTime").as("widnow"))
        val result = table.window(Tumble over 5.second on 'callTime as 'window)
            .groupBy('window, 'sid) // 两个字段分组
            .select('sid, 'window.start, 'window.end, 'sid.count)

        // 输出结果
        tableEnv.toRetractStream[Row](result).filter(_._1 == true).print()

        streamEnv.execute()


        //如果是滑动窗口
        //    table.window(Slide over 10.second every 5.second on 'callTime as 'window)
        //    table.window(Slide.over("10.second").every("5.second").on("callTime").as("window"))
    }
}
